<?php
error_reporting(E_ALL|E_STRICT);
ini_set("display_errors","on");

date_default_timezone_set('Asia/Shanghai');

// 以后入口处可根据此常量确定require哪些文件
define ( 'QUEUE_APP_NAME', 'qieke' );
define ( 'QUEUE_ROOT_DIR', dirname ( dirname ( __FILE__ ) ) );
define ( 'QUEUE_CORE_DIR', QUEUE_ROOT_DIR . '/core' );

require (QUEUE_CORE_DIR . "/queueLog.php");
require (dirname ( dirname ( dirname ( dirname ( __FILE__ ) ) ) ) . '/www/index.php');

set_error_handler('customError');
$queues = include (QUEUE_ROOT_DIR . "/config/apps.cfg.php");

if ($argc != 2) {
	die("error params\n");
}

$queue_name = $argv [1];
if (empty ( $queues [$queue_name] )) {
	die ( "there is no conf for $queue_name\n" );
}

$queue_conf = $queues [$queue_name];

GLOBAL $pids;
$pids = array ();

if (! $queue_conf['run']) {
	exit ();
}

include (QUEUE_ROOT_DIR . "/src/base.queue.php");
include (QUEUE_ROOT_DIR . "/src/$queue_name" . ".queue.php");


setPid ( $queue_name );

$worker_count = $queue_conf['worker_count'];

while ( true )
{
	for($i = 0; $i < $worker_count; ++$i)
	{
		if (! empty ( $pids[$i] ) && posix_kill ( $pids[$i], 0 ))
		{
			//trigger_error("process " . $pids[$i] . " is running");
			continue;
		}
		else
		{
	       // var_dump($pids);
			$pid = pcntl_fork ();
			if ($pid == - 1) {
				die ( "ERROR:fork failed" );
			} else if ($pid) {
				$pids [$i] = $pid;
				trigger_error("new pid=$pid");
				continue;
			} else {
				startApp ( $queue_name, $queue_conf );
				exit ();
			}

		//pcntl_wait();
		}
	}

	declare ( ticks = 1 );
	pcntl_signal ( SIGTERM, "sig_handler" );
	pcntl_signal ( SIGHUP, "sig_handler" );
	pcntl_signal ( SIGINT, "sig_handler" );

	/*
	 * -1		等待任意子进程退出
	 * WNOHANG	如果没有子进程退出立刻返回
	 * pcntl_waitpid 返回0表示没有子进程退出，返回-1表示有错误发生
	 */
	$pid = pcntl_waitpid ( - 1, $status, WNOHANG );
	if ($pid <= 0) {
		// 没有子进程退出
		sleep ( 2 );
	} else {
		trigger_error("process $pid is exit");
	}
}

/**
 * 单个子进程开始执行max_exe_count个任务操作
 * @param $queue_name	队列名称
 * @param $queue_conf	队列的配置信息
 */
function startApp($queue_name, $queue_conf)
{

	$class = $queue_name . "QueueProcessor";
	$processor = new $class ( $queue_name, $queue_conf );

	$max_exe_count = 0;
	if (! isset ( $queue_conf['max_exe_count'] )) {
		$max_exe_count = 1000;
	} else {
		$max_exe_count = $queue_conf['max_exe_count'];
	}

	$processor->process ( $max_exe_count );

	trigger_error("$queue_name task complete");
}

function setPid($queue_name)
{
	$pid = posix_getpid ();
	$ret = file_put_contents ( QUEUE_ROOT_DIR . "/log/monitor/$queue_name.pid", "$pid" );
	if (intval ( $ret ) <= 0) {
		die ( "创建PID文件失败，请检查权限" );
	}
}

function sig_handler($signo)
{
	GLOBAL $pids;
	foreach ( $pids as $pid )
	{
		trigger_error("kill pid $pid");

		if (! posix_kill ( $pid, 0 ))
		{
			trigger_error("process $pid has been exit(zombe)");
			continue;
		}

		$ret = posix_kill ( $pid, SIGINT );
		if ($ret)
		{
			$status = "";
			pcntl_waitpid ( $pid, $status );
			trigger_error("process $pid is exit");
		} else {
			trigger_error("send signal to process $pid failed", E_USER_ERROR);
		}
	}

	exit ();
}

function customError($errno, $errstr, $errfile, $errline)
{
    switch($errno)
    {
        case E_USER_ERROR:
	         QueueLog::errorLog(basename($errfile).'('.$errline.'): '.$errstr);
	         break;
        case E_USER_NOTICE:
        default:
	         QueueLog::debugLog(basename($errfile).'('.$errline.'): '.$errstr);
	         break;
    }
}
